package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.TagInfo
import com.atguigu.userprofile.common.dao.TagInfoDAO
import com.atguigu.userprofile.common.util.ClickhouseUtil.properties
import com.atguigu.userprofile.common.util.{ClickhouseUtil, PropertiesUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.util.Properties

object TaskExport {



  private val properties: Properties =  PropertiesUtil.load("config.properties")
  val HDFS_PATH = properties.getProperty("hdfs-store.path")
  val UP_DBNAME = properties.getProperty("user-profile.dbname")
  val DW_DBNAME = properties.getProperty("data-warehouse.dbname")

  val CLICKHOUSE_URL = properties.getProperty("clickhouse.url")
//0 通过查询标签集合，获得字段名集合
//1  表结构的平移，建一张clickhouse的表 ，结构和hive的表字段一致。
//2   要把数据写入  insert select 不行  scala 在spark 中实现
//如何通过jdbc写入第三方数据库
  def main(args: Array[String]): Unit = {

  val sparkConf: SparkConf = new SparkConf().setAppName("task_export_app").setMaster("local[*]")
  val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()


  val taskId: String = args(0)
    val busiDate: String = args(1)


  //0 通过查询标签集合，获得字段名集合
  val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoList()

  //1  表结构的平移，建一张clickhouse的表 ，结构和hive的表字段一致。
  // clickhouse的建表语句  表名 字段 引擎  主键  排序  分区  颗粒度 8192(可省)
  // create table up_tag_merge_20200614
  // (uid String , 字段名 String ....)
  // engine=MergeTree
  // order by uid

  //  去重  副本（高可用)    去重 =》幂等（实时计算幂等性处理)  批处理不用（企业中批处理的重跑 不用考虑重叠部分，一定是把之前执行失败的残留数据全部清理，完整重跑）

  val tableName=s"up_tag_merge_${busiDate.replace("-","")}"

  val tagCodeList: List[String] = tagInfoList.map(tagInfo => tagInfo.tagCode.toLowerCase + " String")
  val fieldSQL: String = tagCodeList.mkString(",")

  val createTableSQL=
    s"""
       |    create  table  if not exists $UP_DBNAME.$tableName
       |   (uid String ,  $fieldSQL)
       |    engine=MergeTree
       |    order by uid
       |
       |""".stripMargin

  val dropTableSQL=s"drop table if exists $UP_DBNAME.$tableName"


  println(dropTableSQL)
  ClickhouseUtil.executeSql(dropTableSQL)

  println(createTableSQL)
  ClickhouseUtil.executeSql(createTableSQL)

  //2   要把数据写入  insert select 不行  scala 在spark 中实现
  //  hiveTable => dataFrame => jdbc ==> clickhouse

  val dataFrame: DataFrame = sparkSession.sql(s"select * from $UP_DBNAME.$tableName")


  dataFrame.write.mode(SaveMode.Append)
    .option("batchsize",100)  //批量提交 1减少链接 网络io次数 2 减少磁盘碎片小文件
    .option("isolationLevel","NONE")//事务关闭 ck没有事务
    .option("numPartitions","4") //并发
    .option("driver","ru.yandex.clickhouse.ClickHouseDriver")
    .jdbc(CLICKHOUSE_URL,tableName,new Properties())


  }

}
